package lv.solcraft.replication.server.api;

import java.util.Date;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import lv.solcraft.replication.server.bdb.Database;
import lv.solcraft.replication.server.bdb.TxTemplateNoResult;
import lv.solcraft.replication.server.boot.ServerConfig;
import lv.solcraft.replication.server.dao.OperDataAccess;
import lv.solcraft.replication.server.model.Oper;

import org.apache.log4j.Logger;

import com.sleepycat.je.Transaction;

public abstract class AbstractCleanOpersCommand {

    protected static final int BATCH_SIZE = 1000;
    protected final Database database;
    protected OperDataAccess operDataAccess;
    protected final ServerConfig serverConfig;

    public AbstractCleanOpersCommand(Database database, ServerConfig serverConfig) {
        this.database = database;
        this.serverConfig = serverConfig;
        this.operDataAccess = database.getOperDataAccess();
    }

    /**
     * @return next skip count, -1 if there are no more opers to delete
     */
    protected int doDelete(Date expireDate, int skip, int batch) {
        List<Oper> opers = database.getOperDataAccess().findBatchByDateTime(skip, batch);
        getLogger().debug("Processing " + opers.size() + " opers batch");
        int deleted = 0;
        if (opers.isEmpty())
            return -1;
        for (Oper oper : opers) {
            if (oper.getDateTime().compareTo(expireDate) > 0) {
                getLogger().info("Reached opers after expire date, exiting cleaner, oper " + oper);
                return -1;
            }
            if (canDelete(oper)) {
                if (getLogger().isDebugEnabled())
                    getLogger().debug("Deleting oper " + oper);
                deleted++;
                oper.delete();
            } else {
                skip++;
            }
        }
        getLogger().debug("Deleted " + deleted + " opers in batch");
        return skip;
    }

    protected abstract boolean canDelete(Oper oper);

    public final void delete(final Date expireDate) {
        getLogger().info("Starting cleaner, date limit: " + expireDate);
        final AtomicInteger skip = new AtomicInteger(0);
        final AtomicBoolean doIt = new AtomicBoolean(true);
        final AtomicInteger count = new AtomicInteger(0);
        while (doIt.get()) {
            new TxTemplateNoResult(database) {
                @Override
                protected void executeInTxNoResult(Transaction tx) throws Exception {
                    getLogger().debug("Finding opers batch, expire date " + expireDate + ", skip " + skip.get() + ", batch size " + BATCH_SIZE);
                    int nextSkip = doDelete(expireDate, skip.get(), BATCH_SIZE);
                    doIt.set(nextSkip >= 0);
                    skip.set(nextSkip);
                    count.set(count.get() + BATCH_SIZE);
                }
            }.execute();
            if (count.get() >= getLimit()) {
                getLogger().info("More than " + getLimit() + " processed, exiting");
            }
        }
        getLogger().info("Finished cleaner");
    }

    public final void deleteAll() {
        getLogger().info("Deleting all opers");
        final AtomicBoolean doIt = new AtomicBoolean(true);
        while (doIt.get()) {
            new TxTemplateNoResult(database) {
                @Override
                protected void executeInTxNoResult(Transaction tx) throws Exception {
                    List<Oper> opers = operDataAccess.findFirst(BATCH_SIZE);
                    doIt.set(!opers.isEmpty());
                    for (Oper oper : opers) {
                        oper.delete();
                    }
                    getLogger().info("Deleted " + opers.size() + " opers");
                }
            }.execute();
        }
        getLogger().info("Finished to delete all opers");
    }

    protected abstract Logger getLogger();

    protected abstract int getLimit();
}
